package poolResources

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
)

//生产者 ---(生产任务) --->队列 ---(消费任务) --->消费者

// 定义的任务，任务包含需要执行的函数、以及函数要传的参数，因为参数类型、个数不确定，这里适用可变参数核空接口的形式
type Task struct {
	Handle func(v ...interface{})
	Params []interface{}
}

//定义任务池
/*
	任务池的定义包括了
	1、池的容量capacity
	2、当前运行的worker(goroutine)数量runningWorkers
	3、任务队列（channel） chTask以及任务池的状态status（运行中或已关闭，用于安全关闭任务池）
	4、互斥锁
	5、goroutine中产生了panic，会导致整个程序崩溃，为了保证程序的安全进行，任务池对每个work中的panic进行recover操作，并提供可定制的panic handler
*/
type Pool struct {
	capacity       uint64
	runningWorkers uint64
	status         int64
	chTask         chan *Task
	sync.Mutex
	PanicHandler func(interface{})
}

// 任务池的构造函数
var (
	ErrInvalidPoolCap    = errors.New("invalid pool cap")
	ErrPoolAlreadyClosed = errors.New("pool already closed")
)

const (
	RUNNING = 1
	STOPED  = 0
)

func NewPool(capacity uint64) (*Pool, error) {
	if capacity <= 0 {
		return nil, ErrInvalidPoolCap
	}
	return &Pool{
		capacity: capacity,
		status:   RUNNING,
		//初始化任务队列，队列长度为容量
		chTask: make(chan *Task, capacity),
	}, nil
}

/*
runningWorkers的加减直接使用自增运算，如果多个worker时，runningworkers会有数据竞争，这里采用sync.atomic包来保证runningworkers的自增操作原子性
*/
func (p *Pool) incRunning() { //runningWorkers + 1
	atomic.AddUint64(&p.runningWorkers, 1)
}
func (p *Pool) decRunning() {
	atomic.AddUint64(&p.runningWorkers, ^uint64(0))
}
func (p *Pool) GetRunningWorkers() uint64 {
	return atomic.LoadUint64(&p.runningWorkers)
}

/*
对于capacity的操作不需要考虑，在初始化时就已经固化，封装GetCap()方法
*/
func (p *Pool) GetCap() uint64 {
	return p.capacity
}

func (p *Pool) GetWorerNum() uint64 {
	return uint64(len(p.chTask))
}

/*
status的操作也加锁封装，用于安全考虑
*/
func (p *Pool) setStatus(status int64) bool {
	p.Lock()
	defer p.Unlock()
	if p.status == status {
		return false
	}
	p.status = status
	return true
}

/*
启动worker，新建run（）方法作为启动worker的方法
*/
func (p *Pool) run() {
	p.incRunning()
	go func() {
		defer func() {
			p.decRunning()                //worker 结束，运行中的任务减一
			if r := recover(); r != nil { //恢复 panic
				if p.PanicHandler != nil { //如果设置了 PanicHandler，调用}
					p.PanicHandler(r)
				} else { //默认处理
					fmt.Println("worker panic: %s\n", r)
				}
			}
			p.checkWorker() //worker退出时检测是否有可用的worker
		}()
		for {
			select { //阻塞等待任务、结束信号到来
			case task, ok := <-p.chTask: //从channel中消费任务
				if !ok { //如果channel被关闭，结束worker运行
					return
				}
				//执行任务
				task.Handle(task.Params...)
			}
		}
	}()
}

/*
生产任务，新建Put()方法，用来将任务放入池中
*/
func (p *Pool) Put(task *Task) error {
	//加锁防止启动多个worker
	p.Lock()
	defer p.Unlock()
	if p.status == STOPED { // 如果任务池处于关闭状态, 再 put 任务会返回 ErrPoolAlreadyClosed 错误
		return ErrPoolAlreadyClosed
	}
	//run worker
	if p.GetRunningWorkers() < p.GetCap() { //如果任务池满，则不在创建worker
		//创建启动一个worker
		p.run()
	}
	//send task
	if p.status == RUNNING {
		//将任务池推入队列，等待消费
		p.chTask <- task
	}
	return nil
}

/*
任务池安全关闭
当有关闭任务池来节省 goroutine 资源的场景时, 我们需要有一个关闭任务池的方法。
直接销毁 worker 关闭 channel 并不合适, 因为此时可能还有任务在队列中没有被消费掉。要确保所有任务被安全消费后再销毁掉 worker。
首先, 在关闭任务池时, 需要先关闭掉生产任务的入口。同时, 也要考虑到任务推送到 chTask 时 status 改变的问题。改造 Put() 方法:
在 run() 方法中已经对 chTask 的关闭进行了监听, 销毁 worker 只需等待任务被消费完后关闭 chTask。
*/
func (p *Pool) Close() {
	p.setStatus(STOPED) //设置status为已停止
	//for len(p.chTask) == 0 { //阻塞等待所有任务被worker消费
	//	break
	//	time.Sleep(1e6) //防止等待任务情况 cpu负载突然变大，这里小睡一下
	//}
	if len(p.chTask) == 0 {
		close(p.chTask) //关闭任务队列
	}

}

/*
可用worker检查
recover后，goroutine退出，当池容量为1时，此时会有一个问题，当退出时还有运行着的worker
*/
func (p *Pool) checkWorker() {
	p.Lock()
	defer p.Unlock()
	//当前没有worker且有任务存在，运行一个worker消费任务
	//没有任务无需考虑（当前put不会阻塞，下次put会启动worker）
	if p.runningWorkers == 0 && len(p.chTask) > 0 {
		p.run()
	}
}
